package com.abyss.table;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

/**
 * 流形式的SQL开发
 */
public class StreamSQLDemo {
    public static void main(String[] args) throws Exception {
        // 1. 注册table的运行环境
        // 2. 完成将data stream注册为table对象的操作
        // 3. 执行table api完成sql查询
        // 4. 将结果转换回data stream

        /*
        1. 构建执行环境
         */
        // 1.1 Stream的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 流Table的环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        /*
        2. 完成DataStream注册为table的操作
         */
        // 2.1 准备好学生信息数据集
        DataStreamSource<Student> studentDataStreamSource = env.fromElements(
                new Student(1, "潇潇", 11, "female"),
                new Student(2, "刚刚", 10, "male"),
                new Student(3, "蛋蛋", 9, "female"),
                new Student(4, "林俊杰", 8, "male"),
                new Student(5, "林志玲", 9, "female"),
                new Student(6, "刘德华", 8, "male"),
                new Student(7, "张靓颖", 7, "female"),
                new Student(8, "梁朝伟", 11, "male"),
                new Student(9, "欧阳娜娜", 16, "female"),
                new Student(10, "张艺兴", 10, "male"),
                new Student(11, "迪丽热巴", 11, "female")
        );


        // 2.2 构建socket数据流
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);
        SingleOutputStreamOperator<QueryFieldOfId> queryDataStream = socketTextStream.map(new MapFunction<String, QueryFieldOfId>() {
            @Override
            public QueryFieldOfId map(String value) throws Exception {
                return new QueryFieldOfId(Integer.parseInt(value));
            }
        });

        // 2.3 将学生数据集注册为Table
        Table tableStudent = tEnv.fromDataStream(studentDataStreamSource, "id as ida, name, age, sex");

        // 2.4 将查询的id注册为Table
        Table tableQueryOfID = tEnv.fromDataStream(queryDataStream, "id as idb");

        /*
        3. 执行table api完成sql查询
         */
        // 3.1 按照Table API的形式
        // FROM a left join b on a.xx = b.xx
        Table tableResult1 = tableQueryOfID.leftOuterJoin(tableStudent, "ida = idb").select("ida AS id, name, age, sex");

        // 3.2 按照SQL的形式
        // 3.2.1 学生临时表的注册
        tEnv.createTemporaryView("student", studentDataStreamSource, "id, name, age, sex");
        // 3.2.2 查询id的表的注册
        tEnv.createTemporaryView("query_id", queryDataStream, "id");
        // 3.2.3 进行SQL查询
        Table tableResult2 = tEnv.sqlQuery("SELECT s.id, name, age, sex FROM query_id AS q LEFT OUTER JOIN student AS s ON q.id = s.id");

        /*
        4. 将结果转换回data stream
         */
        DataStream<Tuple2<Boolean, Student>> result1 = tEnv.toRetractStream(tableResult1, Student.class);
        DataStream<Tuple2<Boolean, Student>> result2 = tEnv.toRetractStream(tableResult2, Student.class);

        result1.print("result1>>>");
        result2.print("result2>>>");

        env.execute();
    }

    public static class QueryFieldOfId {
        private int id;

        @Override
        public String toString() {
            return "QueryFieldOfId{" +
                    "id=" + id +
                    '}';
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public QueryFieldOfId() {}

        public QueryFieldOfId(int id) {
            this.id = id;
        }
    }

    public static class Student {
        private int id;
        private String name;
        private int age;
        private String sex;

        @Override
        public String toString() {
            return "Student{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", age=" + age +
                    ", sex='" + sex + '\'' +
                    '}';
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }

        public String getSex() {
            return sex;
        }

        public void setSex(String sex) {
            this.sex = sex;
        }

        public Student() {}

        public Student(int id, String name, int age, String sex) {
            this.id = id;
            this.name = name;
            this.age = age;
            this.sex = sex;
        }
    }
}